﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.IO;
using System.Net.Sockets;

using zbus.net;
using zbus.log;
namespace zbus
{
    public class Consumer : MqAdmin, IDisposable
    {
        private static readonly ILogger log = LoggerFactory.GetLogger(typeof(Consumer));
        private MessageClient client = null;
        private string topic = null;
        
        public Consumer(Broker broker, String mq, params MqMode[] modes)
            :base(broker, mq, modes)
        {
        }

        public Consumer(MqConfig config)
            :base(config)
        {
            this.topic = config.topic;
        }

        public string Topic
        {
            set { topic = value;}
        }

        public Message Recv(int timeout)
        {
            if (this.client == null)
            {
                this.client = broker.GetClient(GetClientHint());
            }
            Message req = new Message();
            req.Cmd = Proto.Consume;
            req.Mq = this.mq; 
            if ((this.mode & (int)MqMode.PubSub) != 0)
            {
                if (this.topic != null)
                {
                    req.Topic = this.topic;
                }
            }
            try { 

                Message res = this.client.Invoke(req, timeout);
                if (res != null && res.IsStatus404())
                {
                    if (!this.CreateMQ())
                    {
                        throw new ZbusException("register error");
                    }
                    return Recv(timeout);
                }
                if (res != null )
                {
                    res.Id = res.RawId;
                    res.RemoveHead(Message.RAWID);
                }
                return res;
            }
            catch (IOException ex)
            {
                if (Environment.Version.Major < 4) //.net 3.5 socket sucks!!!!
                {
                    this.HandleFailover();
                }
                else 
                {
                    if (!ex.Message.Contains("period of time")) //timeout just ignore
                    {
                        this.HandleFailover();
                    } 
                }
                
            } 

            return null;

        }

        public void Route(Message msg, int timeout)
        {
            msg.Cmd = Proto.Route;
            msg.Ack = false;
            this.client.Send(msg, timeout);
        }


        private void HandleFailover()
        {
            try
            {
                broker.CloseClient(this.client);
                this.client = broker.GetClient(GetClientHint());
            }
            catch (IOException ex)
            {
                log.Error(ex.Message, ex);
            }
            
        }

        public void Dispose()
        {
            if (this.client != null)
            {
                broker.CloseClient(this.client);
            }
        }
    }
}
